package glive.system.peer.glive;

import common.configuration.CyclonConfiguration;
import common.configuration.GLiveConfiguration;
import common.peer.DescriptorBuffer;
import common.peer.PeerAddress;
import common.peer.PeerDescriptor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.CancelTimeout;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import common.peer.PeerData;
import common.configuration.Global;
import common.events.PeerDataRequest;
import common.events.PeerDataResponse;
import common.simulation.PeerStatus;
import common.web.WebApplication;
import cyclon.system.peer.cyclon.CyclonPartnersPort;
import cyclon.system.peer.cyclon.CyclonPartnersRequest;
import cyclon.system.peer.cyclon.CyclonPartnersResponse;

public final class GLive extends ComponentDefinition {

    Negative<PeerStatus> peerStatusPort = negative(PeerStatus.class);
    Positive<CyclonPartnersPort> cyclonPartnersPort = positive(CyclonPartnersPort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
	//private PeerDescriptor self;
	private HashMap<String, Integer> eventCounts;
	private long shuffleTimeout;
	private Cache cache;
	private boolean joining;
	Logger logger;
	long period;
	private GLiveConfiguration gliveConfiguration;
	private HashMap<UUID, PeerAddress> outstandingRandomShuffles;

//-------------------------------------------------------------------	
	public GLive() {
		outstandingRandomShuffles = new HashMap<UUID, PeerAddress>();
		eventCounts = new HashMap<String, Integer>();

        subscribe(handleInit, control);
        subscribe(handleRound, timerPort);
        subscribe(handleBufferTimeout, timerPort);
        subscribe(handleCyclonPartnersResponse, cyclonPartnersPort);
        subscribe(handleGLiveBufferRequest, networkPort);
        subscribe(handleGLiveBufferResponse, networkPort);
        subscribe(handleAcquireParent, networkPort);
        subscribe(handleParentResponse, networkPort);
        subscribe(handleUpdateCost, networkPort);
        subscribe(handleOrphanSelf, networkPort);
        subscribe(handleOrphanChild, networkPort);
        subscribe(handlePeerDataRequest, peerStatusPort);
	}

//-------------------------------------------------------------------	
	Handler<GLiveInit> handleInit = new Handler<GLiveInit>() {
		public void handle(GLiveInit init) {
			logger = LoggerFactory.getLogger(GLive.class
					.getName()
					+ "@" + init.getSelf());
			logger.debug("INIT");
			joining = true;
			//self = new PeerDescriptor(init.getSelf());
			cache = new Cache(Global.tvs, logger, new PeerDescriptor(init.getSelf()));
            gliveConfiguration = init.getConfiguration();
            period = gliveConfiguration.getPeriod();
            //Set the shuffle timeout with any arbitrary high value
            shuffleTimeout = 50000;
            SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
            rst.setTimeoutEvent(new GLiveSchedule(rst));
            trigger(rst, timerPort);

    }
};

//-------------------------------------------------------------------  
Handler<GLiveSchedule> handleRound = new Handler<GLiveSchedule>() {
    public void handle(GLiveSchedule event) {
            CyclonPartnersRequest response = new CyclonPartnersRequest();
            if(joining)
            	response.setSelf(cache.getSelf());
            trigger(response, cyclonPartnersPort);
    }
};

//-------------------------------------------------------------------Edited
Handler<PeerDataRequest> handlePeerDataRequest = new Handler<PeerDataRequest>() {
	public void handle(PeerDataRequest event) {
		PeerData pd = new PeerData(cache.getSelf().getPeerAddress(), toPeerAddressList(cache.getParents()), toPeerAddressList(cache.getChildren()), toPeerAddressList(cache.getAllCyclonPartners()), toPeerAddressList(cache.getAllSimilarEntries()), cache.getAverageCyclonAge(), cache.getAverageSimilarAge(), cache.getSelf().getCost());
		pd.setEventCounts(eventCounts);
		pd.setSp(cache.getSelf().getSP());
		pd.setFingersList(fromMapToPeerAddressList(cache.getAllFingers()));

		for(String eventType: eventCounts.keySet())
		{
			eventCounts.put(eventType, 0);
		}
		PeerDataResponse responseEvent = new PeerDataResponse(event, pd);
		trigger(responseEvent, peerStatusPort);
	}
};

//-------------------------------------------------------------------  
Handler<CyclonPartnersResponse> handleCyclonPartnersResponse = new Handler<CyclonPartnersResponse>() {
        public void handle(CyclonPartnersResponse event) {
        		//Increment Age of Similar View entries
        		cache.incrementDescriptorAges();  
    			//System.out.println(event.getPartners());
        		//Merge Cyclon View       		
        		cache.MergeCyclonPartners(event.getPartners(), joining);
        		if(joining == true)
        			joining = false;
        		        		     		
        		//Schedule GLive Buffer Request
    			PeerAddress randomPeer = cache.selectPeerToShuffleWith();
    			//But only if there's at least one peer available
    			//randomPeer = null;
    			if(randomPeer != null)
    			{
	    			//Shuffle size is set to be the same as that of Cyclon [CR]
	        		ArrayList<PeerDescriptor> randomDescriptors = cache.selectToSendAtActive(Global.tsl - 1, randomPeer);
	        		randomDescriptors.add(cache.getSelf().copyPD());
	        		DescriptorBuffer randomBuffer = new DescriptorBuffer(cache.getSelf().getPeerAddress(), randomDescriptors);
	        		//Timeout if peer doesn't respond back
	        		ScheduleTimeout rst = new ScheduleTimeout(shuffleTimeout);
	        		rst.setTimeoutEvent(new GLiveBufferRequestTimeout(rst, randomPeer));
	        		UUID rTimeoutId = rst.getTimeoutEvent().getTimeoutId();
	        		outstandingRandomShuffles.put(rTimeoutId, randomPeer);
	        		GLiveBufferRequest rRequest = new GLiveBufferRequest(rTimeoutId, randomBuffer, cache.getSelf().getPeerAddress(), randomPeer);
	        		trigger(rst, timerPort);
	        		trigger(rRequest, networkPort);
    			}
        		

        }
};

//-------------------------------------------------------------------	
	Handler<GLiveBufferRequest> handleGLiveBufferRequest = new Handler<GLiveBufferRequest>() {
		public void handle(GLiveBufferRequest event) {
			//Do nothing if I'm still in the process of joining (i.e. my similar view is still being boostrapped)
			if(!joining)
			{
				PeerAddress peer = event.getPeerSource();
				DescriptorBuffer receivedSimilarBuffer = event.getRandomBuffer();
				DescriptorBuffer toSendSimliarBuffer = new DescriptorBuffer(cache.getSelf().getPeerAddress(), cache.selectToSendAtPassive(receivedSimilarBuffer.getSize(), peer));
				//Acquire parents if necessary
				ArrayList<PeerDescriptor> peersToAcquire = cache.selectToKeep(peer, receivedSimilarBuffer.getDescriptors());
				for(PeerDescriptor pta: peersToAcquire)
				{
					countEvent("AcquireParent");
					GLiveAcquireParent request = new GLiveAcquireParent(cache.getSelf().getPeerAddress(), pta.getPeerAddress(), cache.getSelf().copyPD());
					trigger(request, networkPort);
				}
				GLiveBufferResponse response = new GLiveBufferResponse(event.getRequestId(), toSendSimliarBuffer, cache.getSelf().getPeerAddress(), peer);
				trigger(response, networkPort);
			}
		}
	};

//-------------------------------------------------------------------	
	Handler<GLiveBufferResponse> handleGLiveBufferResponse = new Handler<GLiveBufferResponse>() {
		public void handle(GLiveBufferResponse event) {

			// cancel shuffle timeout
			UUID shuffleId = event.getRequestId();
			if (outstandingRandomShuffles.containsKey(shuffleId)) {
				outstandingRandomShuffles.remove(shuffleId);
				CancelTimeout ct = new CancelTimeout(shuffleId);
				trigger(ct, timerPort);
			}

			PeerAddress peer = event.getPeerSource();
			DescriptorBuffer receivedRandomBuffer = event.getRandomBuffer();
			ArrayList<PeerDescriptor> peersToAcquire = cache.selectToKeep(peer, receivedRandomBuffer.getDescriptors());
			for(PeerDescriptor pta: peersToAcquire)
			{
				countEvent("AcquireParent");
				GLiveAcquireParent request = new GLiveAcquireParent(cache.getSelf().getPeerAddress(), pta.getPeerAddress(), cache.getSelf().copyPD());
				trigger(request, networkPort);
			}
		}
	};
	
//------------------------------------------------------------------
	Handler<GLiveAcquireParent> handleAcquireParent = new Handler<GLiveAcquireParent>() {
		public void handle(GLiveAcquireParent event) {
			boolean decision = cache.decideOnChild(event.getChild());
			logger.debug("Peer {} wants to acquire me", event.getChild().getPeerAddress());
			if(decision)
			{
				logger.debug("I've decided to accept peer {} as my child", event.getChild().getPeerAddress());
				GLiveParentResponse response = new GLiveParentResponse(cache.getSelf().getPeerAddress(), event.getChild().getPeerAddress(), cache.getSelf().copyPD(), true);
				trigger(response, networkPort);
				// Check & Remove unwanted children
				PeerAddress childToBeRemoved = cache.getChildToBeRemoved();
				if(childToBeRemoved != null)
				{
					logger.debug("I've decided to orphan my child {}", childToBeRemoved);
					countEvent("OrphanChild");
					GLiveOrphanChild notify = new GLiveOrphanChild(cache.getSelf().getPeerAddress(), childToBeRemoved, cache.getSelf().copyPD());
					trigger(notify, networkPort);
				}
			}
			else
			{
				logger.debug("I've decided to reject peer {} to become my child", event.getChild().getPeerAddress());
				GLiveParentResponse response = new GLiveParentResponse(cache.getSelf().getPeerAddress(), event.getChild().getPeerAddress(), cache.getSelf().copyPD(), false);
				trigger(response, networkPort);
			}
		}
	};
	
//------------------------------------------------------------------
	Handler<GLiveParentResponse> handleParentResponse = new Handler<GLiveParentResponse>() {
		public void handle(GLiveParentResponse event) {
			if(event.getResponse())
			{
				logger.debug("Peer {} accepts my acquisition", event.getParent().getPeerAddress());
				boolean decision = cache.stillWantParent(event.getParent());		
				//Accept parent
				if(decision)
				{
					logger.debug("I've decided to accept peer {}:{} as my parent", event.getParent().getPeerAddress(), event.getParent().getMoney());
					//Check & Remove unwanted parents
					PeerAddress parentToBeRemoved = cache.getParentToBeRemoved();
					if(parentToBeRemoved != null)
					{
						logger.debug("I've decided to remove parent {}:{}", parentToBeRemoved, parentToBeRemoved.getBandwidth());
						countEvent("OrphanSelf");
						GLiveOrphanSelf notify = new GLiveOrphanSelf(cache.getSelf().getPeerAddress(), parentToBeRemoved, cache.getSelf().copyPD());
						trigger(notify, networkPort);
					}
					if(cache.hasCostChanged())
					{
						//Send Cost Update to children (if any exist)
						for(PeerDescriptor child: cache.getChildren())
						{
							countEvent("UpdateCost");
							GLiveUpdateCost updateNotification = new GLiveUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
							trigger(updateNotification, networkPort);
						}
					}
				}
				//Reject parent
				else
				{
					logger.debug("I've decided to reject parent {}'s offer", event.getParent().getPeerAddress());
					countEvent("OrphanSelf");
					GLiveOrphanSelf notify = new GLiveOrphanSelf(cache.getSelf().getPeerAddress(), event.getParent().getPeerAddress(), cache.getSelf().copyPD());
					trigger(notify, networkPort);
				}
			}
			else
			{
				logger.debug("Peer {} rejects my acquisition", event.getParent().getPeerAddress());
				cache.updateViewEntry(event.getParent());
			}
		}
	};
	
//-------------------------------------------------------------------
	Handler<GLiveOrphanChild> handleOrphanChild = new Handler<GLiveOrphanChild>() {
		public void handle (GLiveOrphanChild event) {
			logger.debug("Peer {} decided to orphan me", event.getParent().getPeerAddress());
			cache.removeParent(event.getParent());
			//Check if my Path length to the source has changed
			if(cache.hasCostChanged())
			{
				//Send Cost Update to children (if any exist)
				for(PeerDescriptor child: cache.getChildren())
				{
					countEvent("UpdateCost");
					GLiveUpdateCost updateNotification = new GLiveUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
					trigger(updateNotification, networkPort);
				}
			}
		}
	};
//-------------------------------------------------------------------
	Handler<GLiveOrphanSelf> handleOrphanSelf = new Handler<GLiveOrphanSelf>() {
		public void handle (GLiveOrphanSelf event) {
			logger.debug("Child {} relinquished its relationship with me", event.getChild().getPeerAddress());
			cache.removeChild(event.getChild());
		}
	};
	
//-------------------------------------------------------------------
	Handler<GLiveUpdateCost> handleUpdateCost = new Handler<GLiveUpdateCost>() {
		public void handle (GLiveUpdateCost event) {
			cache.updateParent(event.getParent());
			if(cache.hasCostChanged())
			{
				//Send Cost Update to children (if any exist)
				for(PeerDescriptor child: cache.getChildren())
				{
					countEvent("UpdateCost");
					GLiveUpdateCost updateNotification = new GLiveUpdateCost(cache.getSelf().getPeerAddress(), child.getPeerAddress(), cache.getSelf().copyPD());
					trigger(updateNotification, networkPort);
				}
			}
		}
	};
//-------------------------------------------------------------------	
	Handler<GLiveBufferRequestTimeout> handleBufferTimeout = new Handler<GLiveBufferRequestTimeout>() {
		public void handle(GLiveBufferRequestTimeout event) {
			//This handler should remove the sent-to entries for the target peer from the cache
			//throw new RuntimeException("BufferRequest Timed-out in an unexpected manner");
			logger.info("Unexpected BufferTimeout");
		}
	};
	
//-------------------------------------------------------------------
	private ArrayList<PeerDescriptor> getPartners() {
		ArrayList<PeerDescriptor> partnersDescriptors = cache.getAllCyclonPartners();
		ArrayList<PeerDescriptor> partners = new ArrayList<PeerDescriptor>();
		for (PeerDescriptor desc : partnersDescriptors)
		{
			partners.add(desc);
		}
		
		return partners;
	}
	
	private void countEvent(String eventName)
	{
		if(eventCounts.get(eventName)==null)
			eventCounts.put(eventName, 1);
		else eventCounts.put(eventName, eventCounts.get(eventName)+1);
	}
	
	private ArrayList<PeerAddress> toPeerAddressList(ArrayList<PeerDescriptor> map)
	{
		ArrayList<PeerAddress> myList = new ArrayList<PeerAddress>();
		for (PeerDescriptor desc : map)
		{
			myList.add(desc.getPeerAddress());
		}
		
		return myList;	
	}
	
	private ArrayList<PeerAddress> fromMapToPeerAddressList(HashMap<Integer, PeerDescriptor> map)
	{
		ArrayList<PeerAddress> myList = new ArrayList<PeerAddress>();
		for (Integer BW : map.keySet())
		{
			myList.add(map.get(BW).getPeerAddress());
		}
		
		return myList;	
	}
}
